package cn.dmrliu.edu.realtime.app.func;

import cn.dmrliu.edu.realtime.common.EduConfig;
import cn.dmrliu.edu.realtime.util.PhoenixUtil;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.util.Collection;
import java.util.Set;


public class DimSinkFunction implements SinkFunction<JSONObject> {
    @Override
    public void invoke(JSONObject jsonObject, Context ctx) throws Exception {
        String sinkTable = jsonObject.getString("sinkTable");
        jsonObject.remove("sinkTable");

        Set<String> keySet = jsonObject.keySet();
        Collection<Object> values = jsonObject.values();

        StringBuffer upsertSql = new StringBuffer();
        upsertSql
                .append("upsert into ").append(EduConfig.PHOENIX_SCHEMA).append(".").append(sinkTable)
                .append("(").append(StringUtils.join(keySet, ",")).append(")")
                .append(" values ('").append(StringUtils.join(values, "','")).append("')");

        System.out.println("phoenix插入语句是 = " + upsertSql);

        PhoenixUtil.executeSql(upsertSql.toString());
    }
}
